package mspbots.smileback.client.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Builder;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.smileback.client.oauth.AuthorizationCode;
import mspbots.smileback.common.RabbitKeys;
import mspbots.smileback.common.RedisKeys;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Map;

/**
 * mspbots.data.teams.core.Crontab
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/6/3
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class GlobalScheduledTimer {

    private final Map<String, BaseSyncService> syncServiceMap;

    private final RabbitMessagingTemplate rabbitMessagingTemplate;
    private final ReactiveStringRedisTemplate stringRedisTemplate;
    private final ReactiveRedisTemplate<String, Object> reactiveRedisTemplate;
    private final ObjectMapper objectMapper;

    /**
     * 测试替换你开发的服务即可
     * (v instanceof ProjectsServiceImpl)
     * 修改请勿提交此文件
     * <code>@PostConstruct</code>
     */
    @PostConstruct
    public void autoSyncTest() {
        this.reactiveRedisTemplate.scan(ScanOptions.scanOptions()
                .match(RedisKeys.OAUTH_TENANT_REDIS_KEY_PREFIX + "*").count(100).build())
                .flatMap(key -> this.reactiveRedisTemplate.opsForValue().get(key))
                .map(data -> this.objectMapper.convertValue(data, AuthorizationCode.class))
                .map(AuthorizationCode::getTenantId)
                .subscribe();
    }

    @Scheduled(fixedDelay = 1000 * 60)
    public void autoSyncFast() {
        DistributedTask distributedTask = DistributedTask.builder().type("fast").build();
        this.distributedTaskLock(distributedTask)
                .doFinally(r -> this.stringRedisTemplate.opsForValue()
                        .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType(),
                                "false", Duration.ofMinutes(1)).subscribe())
                .block(Duration.ofSeconds(10));
    }

    @Scheduled(fixedDelay = 1000 * 3600 * 5)
    public void autoSyncSlow() {
        DistributedTask distributedTask = DistributedTask.builder().type("slow").build();
        this.distributedTaskLock(distributedTask)
                .doFinally(r -> this.stringRedisTemplate.opsForValue()
                        .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType(),
                                "false", Duration.ofHours(5)).subscribe())
                .block(Duration.ofSeconds(10));
    }

    private Mono<Void> distributedTaskLock(DistributedTask distributedTask) {
        return this.stringRedisTemplate.opsForValue()
                .get(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType())
                .map(Boolean::parseBoolean).defaultIfEmpty(true)
                .filter(r -> r)
                .flatMapMany(r -> this.onSendTask(distributedTask))
                .doOnNext(tenant -> log.debug("Sync send distributed [{}] ,tenant [{}]",
                        distributedTask.getType(), tenant))
                .then();
    }

    private Flux<Long> onSendTask(DistributedTask distributedTask) {
        return this.reactiveRedisTemplate.scan(ScanOptions.scanOptions()
                .match(RedisKeys.OAUTH_TENANT_REDIS_KEY_PREFIX + "*").count(100).build())
                .flatMap(key -> this.reactiveRedisTemplate.opsForValue().get(key))
                .map(data -> this.objectMapper.convertValue(data, AuthorizationCode.class))
                .filterWhen(data -> this.stringRedisTemplate.opsForValue()
                        .size(RedisKeys.ON_OFF_TENANT_TASK_REDIS_KEY + data.getTenantId())
                        .defaultIfEmpty(0L).map(r -> r == 0))
                .map(AuthorizationCode::getTenantId)
                .doOnNext(tokenState -> this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                        RabbitKeys.DISTRIBUTED_TASK_QUEUE_KEY, distributedTask.setTenantId(tokenState)));
    }

    @RabbitListener(queues = "smile-back.distributed.task.sync", concurrency = "10")
    public void syncTask(DistributedTask distributedTask) {

        Long tenantId = distributedTask.getTenantId();

        String lockKey = RedisKeys.TENANT_RUNNING_LOCK_ANCHOR_REDIS_KEY
                + distributedTask.getType() + ":" + tenantId;

        this.stringRedisTemplate.opsForValue().size(lockKey)
                .filter(count -> count == 0)
                .doOnNext(count -> this.reactiveRedisTemplate.opsForValue()
                        .setIfAbsent(lockKey, tenantId, Duration.ofMinutes(60))
                        .subscribe())
                .flatMapMany(count -> Flux.fromStream(this.syncServiceMap.keySet().parallelStream()))
                .flatMap(key -> {
                    log.debug("===Sync [{}] {} Task [{}] Running===", tenantId, distributedTask.getType(), key);
                    if ("fast".equals(distributedTask.getType())) {
                        return this.syncServiceMap.get(key).syncFast(tenantId)
                                .onErrorResume(err -> {
                                    log.error("Sync [{}] task [{}]  error, msg: [{}]",
                                            tenantId, key, err.getMessage());
                                    return Mono.empty();
                                });
                    } else {
                        return this.syncServiceMap.get(key).syncSlow(tenantId)
                                .onErrorResume(err -> {
                                    log.error("Sync [{}] task [{}]  error, msg: [{}]",
                                            tenantId, key, err.getMessage());
                                    return Mono.empty();
                                });
                    }
                })
                .doFinally(key -> this.reactiveRedisTemplate.opsForValue().delete(lockKey).subscribe())
                .subscribe();
    }

    @Data
    @Builder
    static class DistributedTask {
        private String type;
        private Long tenantId;

        public DistributedTask setTenantId(Long tenantId) {
            this.tenantId = tenantId;
            return this;
        }

    }
}
